package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetailApp extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeOrderDetailApp().init(2007,"DwdTradeOrderDetailApp");
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        tEnv.executeSql("create table dwd_trade_order_app(" +
                " id string, " +
                " course_id string, " +
                " course_name string, " +
                " order_id string, " +
                " user_id string, " +
                " origin_amount string, " +
                " coupon_reduce string, " +
                " final_amount string, " +
                " session_id string, " +
                " create_time string, " +
                " update_time string, " +
                " order_status string, " +
                " out_trade_no string, " +
                " trade_body string, " +
                " province_id string, " +
                " expire_time string, " +
                " od_ts string, " +
                " xid string, " +
                " `old` map<string,string>, " +
                " `type` string, " +
                " oi_ts string" +
                ")" + SqlUtil.getKafkaReadDDL(Constant.TOPIC_DWD_TRADE_ORDER_APP, "DwdTradeOrderDetailApp"));

        Table resultTable = tEnv.sqlQuery("select " +
                " id, " +
                " course_id, " +
                " course_name, " +
                " order_id, " +
                " user_id, " +
                " origin_amount, " +
                " coupon_reduce, " +
                " final_amount, " +
                " session_id, " +
                " create_time, " +
                " update_time, " +
                " order_status, " +
                " out_trade_no, " +
                " trade_body, " +
                " province_id, " +
                " expire_time, " +
                " od_ts ts " +
                "from dwd_trade_order_app " +
                "where `type`='insert'");
        //resultTable.printSchema();

        tEnv.executeSql("create table dwd_trade_order_detail_app(" +
                " id string, " +
                " course_id string, " +
                " course_name string, " +
                " order_id string, " +
                " user_id string, " +
                " origin_amount string, " +
                " coupon_reduce string, " +
                " final_amount string, " +
                " session_id string, " +
                " create_time string, " +
                " update_time string, " +
                " order_status string, " +
                " out_trade_no string, " +
                " trade_body string, " +
                " province_id string, " +
                " expire_time string, " +
                " ts string " +
                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL_APP));

        resultTable.executeInsert("dwd_trade_order_detail_app");


    }
}